package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.article.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * @作者 itcast
 * @创建日期 2021/4/26 14:28
 **/
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        // 流计算的处理逻辑   key: uuid      value: {type: LIKES, articleId:123}
        return stream.map((key,value)->{
            log.info("流处理器接收到数据==> key:{}    ,  value:{}",key,value);
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);
            //将数据封装成新的key   key的格式:  文章id:操作类型   123:LIKES
            return new KeyValue<>(updateArticleMess.getArticleId()+":"+updateArticleMess.getType().name(),key);
        }).groupByKey() // 按照key进行分组  可以得到每个文章如 点赞的数量统计
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10))) // 时间窗口，每隔多久统计一次
                .count() // 统计key出现的次数
                .toStream()  // 将统计的结果输出到新的流中  key: 对象    value: 个数
                .map((key,value)->{
                    //  将统计结果封装成一个对象  key.key() :   123:LIKES     value:  4
                    ArticleVisitStreamMess streamMess = formatObj(key.key(),value);
                    log.info("经过流计算得到的统计结果===============>{}",streamMess);
                    // 将结果消息发送到新的主题中
                    return new KeyValue<>(key.key(),JSON.toJSONString(streamMess));
                });



    }

    private ArticleVisitStreamMess formatObj(String key, Long value) {//  123:LIKES     value:  4
        String[] articleIdAndType = key.split(":");
        String articleId = articleIdAndType[0];
        String typeName = articleIdAndType[1];
        ArticleVisitStreamMess streamMess = new ArticleVisitStreamMess();
        streamMess.setArticleId(Long.valueOf(articleId));
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())){
            streamMess.setView(value);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())){
            streamMess.setCollect(value);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())){
            streamMess.setComment(value);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.LIKES.name())){
            streamMess.setLike(value);
        }
        return streamMess;
    }
}
